// go-actor-framework-full.go
// 完整的 Go Actor 框架，包含监督者、Actor 池、自动扩缩容、优雅关闭、强制停止和 Prometheus 指标

package main

import (
	"context"
	"fmt"
	"net/http"
	"os"
	"os/signal"
	"runtime/debug"
	"sync"
	"syscall"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promhttp"
)

// ---------------- Metrics ----------------
// Actor 队列长度指标
var (
	actorQueueLength = prometheus.NewGaugeVec(
		prometheus.GaugeOpts{Name: "actor_queue_length", Help: "每个 actor 当前待处理消息数"},
		[]string{"actor"},
	)
	// Actor 池大小指标
	actorPoolSize = prometheus.NewGaugeVec(
		prometheus.GaugeOpts{Name: "actor_pool_size", Help: "当前 actor 池大小"},
		[]string{"pool"},
	)
	// Actor 重启次数指标
	actorRestarts = prometheus.NewCounterVec(
		prometheus.CounterOpts{Name: "actor_restarts_total", Help: "actor 重启次数"},
		[]string{"actor"},
	)
)

func init() {
	// 注册 Prometheus 指标
	prometheus.MustRegister(actorQueueLength, actorPoolSize, actorRestarts)
}

// ---------------- Core Types ----------------

// 消息类型，任意类型都可作为消息
type Message = interface{}

// Actor 重启策略类型
type RestartStrategy int

// Actor 重启策略枚举
const (
	RestartNever     RestartStrategy = iota // 永不重启
	RestartAlways                           // 总是重启
	RestartOnFailure                        // 失败时重启
)

// Actor 属性配置
// ActorFunc: 处理消息的函数
// Mailbox: 邮箱容量
// Strategy: 重启策略
// MaxRestarts: 最大重启次数
// Window: 重启计数窗口
type Props struct {
	ActorFunc   func(ctx Context, msg Message)
	Mailbox     int
	Strategy    RestartStrategy
	MaxRestarts int
	Window      time.Duration
}

// Actor 引用，包含路径、消息通道、关闭信号
type ActorRef struct {
	path string
	send chan<- Message
	done <-chan struct{}
}

// 向 Actor 发送消息
func (r ActorRef) Tell(msg Message) {
	select {
	case <-r.done:
		fmt.Printf("[WARN] actor %s 正在关闭，丢弃消息 %#v\n", r.path, msg)
	case r.send <- msg:
	}
}

// 获取 Actor 路径
func (r ActorRef) Path() string { return r.path }

// Actor 上下文，包含自身引用、状态、系统指针、停止信号
type Context struct {
	Self  ActorRef
	state map[string]interface{}
	sys   *actorSystem
	stop  chan struct{}
}

// 创建子 Actor
func (c *Context) SpawnChild(name string, props Props) ActorRef {
	return c.sys.spawnChild(name, props, &c.Self)
}

// 停止 Actor
func (c *Context) Stop() { close(c.stop) }

// ---------------- Actor System ----------------

// Actor 系统，管理所有子 Actor
type actorSystem struct {
	mu       sync.Mutex
	children map[string]*child
	wg       sync.WaitGroup
}

// 子 Actor 结构体
// name: 名称
// props: 属性
// mailbox: 消息队列
// ref: Actor 引用
// stop: 停止信号
// closing: 关闭信号
// restarts: 重启时间记录
type child struct {
	name     string
	props    Props
	mailbox  chan Message
	ref      ActorRef
	stop     chan struct{}
	closing  chan struct{}
	mu       sync.Mutex
	restarts []time.Time
}

// 获取待处理消息数
func (c *child) pending() int { return len(c.mailbox) }

// 更新队列长度指标
func (c *child) updateMetrics() {
	actorQueueLength.WithLabelValues(c.name).Set(float64(len(c.mailbox)))
}

// 增加重启次数指标
func (c *child) incrementRestartMetric() { actorRestarts.WithLabelValues(c.name).Inc() }

// 创建 Actor 系统
func newSystem() *actorSystem { return &actorSystem{children: make(map[string]*child)} }

// 创建 Actor
func (s *actorSystem) Spawn(name string, props Props) ActorRef { return s.spawnChild(name, props, nil) }

// 创建子 Actor，支持父子关系
func (s *actorSystem) spawnChild(name string, props Props, parent *ActorRef) ActorRef {
	if props.Mailbox <= 0 {
		props.Mailbox = 16 // 默认邮箱容量
	}
	if props.Window == 0 {
		props.Window = 5 * time.Second // 默认重启窗口
	}
	c := &child{name: name, props: props, mailbox: make(chan Message, props.Mailbox), stop: make(chan struct{}), closing: make(chan struct{})}
	c.ref = ActorRef{path: name, send: c.mailbox, done: c.closing}

	// 加入系统管理
	s.mu.Lock()
	if _, exists := s.children[name]; exists {
		s.mu.Unlock()
		panic("actor name already exists: " + name)
	}
	s.children[name] = c
	s.mu.Unlock()

	// 启动 Actor
	s.runChild(c)
	return c.ref
}

// 启动子 Actor 的主循环，处理消息、异常与重启
func (s *actorSystem) runChild(c *child) {
	s.wg.Add(1)
	go func() {
		defer s.wg.Done()
		for {
			// 每次重启都新建 Context
			ctx := Context{Self: c.ref, state: make(map[string]interface{}), sys: s, stop: make(chan struct{})}
			exit := make(chan interface{})
			go func() {
				defer func() {
					if r := recover(); r != nil {
						exit <- r // 捕获异常
						return
					}
					exit <- nil // 正常退出
				}()
				for {
					select {
					case msg := <-c.mailbox:
						c.props.ActorFunc(ctx, msg) // 处理消息
					case <-ctx.stop:
						return // 主动停止
					case <-c.stop:
						return // 强制停止
					case <-c.closing:
						// 关闭时清空队列
					drain:
						for {
							select {
							case msg := <-c.mailbox:
								c.props.ActorFunc(ctx, msg)
							default:
								break drain
							}
						}
						return
					}
				}
			}()

			res := <-exit
			if res == nil {
				// 正常退出，移除 Actor
				s.mu.Lock()
				delete(s.children, c.name)
				s.mu.Unlock()
				return
			}

			// 异常退出，打印错误和堆栈
			fmt.Printf("[ERROR] actor %s 崩溃: %v\nStack:\n%s\n", c.name, res, string(debug.Stack()))
			c.incrementRestartMetric()
			c.mu.Lock()
			now := time.Now()
			c.restarts = append(c.restarts, now)
			cut := now.Add(-c.props.Window)
			j := 0
			for i := len(c.restarts) - 1; i >= 0; i-- {
				if c.restarts[i].After(cut) {
					j = i
				}
			}
			c.restarts = c.restarts[j:]
			count := len(c.restarts)
			c.mu.Unlock()

			// 判断是否重启
			shouldRestart := false
			switch c.props.Strategy {
			case RestartAlways:
				shouldRestart = true
			case RestartOnFailure:
				if count <= c.props.MaxRestarts || c.props.MaxRestarts == 0 {
					shouldRestart = true
				}
			}

			if !shouldRestart {
				fmt.Printf("[SUPERVISOR] 不重启 %s\n", c.name)
				s.mu.Lock()
				delete(s.children, c.name)
				s.mu.Unlock()
				return
			}

			// 重启退避时间，防止频繁重启
			backoff := time.Millisecond * 100 * time.Duration(count)
			if backoff > 5*time.Second {
				backoff = 5 * time.Second
			}
			fmt.Printf("[SUPERVISOR] %s %v 后重启 (count=%d)\n", c.name, backoff, count)
			time.Sleep(backoff)
		}
	}()
}

// 优雅关闭 Actor 系统
func (s *actorSystem) Shutdown(ctx context.Context) error {
	s.mu.Lock()
	for _, c := range s.children {
		select {
		case <-c.closing:
		default:
			close(c.closing)
		}
	}
	s.mu.Unlock()

	done := make(chan struct{})
	go func() { s.wg.Wait(); close(done) }()

	select {
	case <-done:
		return nil // 正常关闭
	case <-ctx.Done():
		// 超时后强制停止所有 Actor
		s.mu.Lock()
		for _, c := range s.children {
			select {
			case <-c.stop:
			default:
				close(c.stop)
			}
		}
		s.mu.Unlock()
		return ctx.Err()
	}
}

// ---------------- ActorPool + AutoScaler ----------------

// 自动扩缩容配置
type AutoScalerConfig struct {
	Interval      time.Duration
	HighThreshold int
	LowThreshold  int
	ScaleUpStep   int
	ScaleDownStep int
	Cooldown      time.Duration
}

// Actor 池，管理一组 worker Actor
type ActorPool struct {
	name     string
	sys      *actorSystem
	props    Props
	min, max int
	workers  []*child
	mu       sync.Mutex
	next     int
	scaling  struct {
		cfg  AutoScalerConfig
		stop chan struct{}
		cool time.Time
	}
}

func NewPool(sys *actorSystem, name string, props Props, min, max int) *ActorPool {
	if min < 1 {
		min = 1 // 最小池大小
	}
	if max < min {
		max = min // 最大池不能小于最小池
	}
	p := &ActorPool{name: name, sys: sys, props: props, min: min, max: max}
	p.Resize(min)
	return p
}

// 调整池大小
func (p *ActorPool) Resize(n int) {
	p.mu.Lock()
	defer p.mu.Unlock()
	if n < p.min {
		n = p.min
	}
	if n > p.max {
		n = p.max
	}
	cur := len(p.workers)
	if n > cur {
		for i := cur; i < n; i++ {
			name := fmt.Sprintf("%s-worker-%d", p.name, i)
			p.sys.spawnChild(name, p.props, nil)
			child := p.sys.children[name]
			p.workers = append(p.workers, child)
		}
		fmt.Printf("[POOL] 扩容 %s: %d -> %d\n", p.name, cur, n)
	} else if n < cur {
		p.workers = p.workers[:n]
		fmt.Printf("[POOL] 缩容 %s: %d -> %d\n", p.name, cur, n)
	}
}

// 向池中的 worker 轮询发送消息
func (p *ActorPool) Tell(msg Message) {
	p.mu.Lock()
	if len(p.workers) == 0 {
		p.mu.Unlock()
		fmt.Printf("[POOL] %s 没有 worker\n", p.name)
		return
	}
	w := p.workers[p.next%len(p.workers)]
	p.next = (p.next + 1) % len(p.workers)
	p.mu.Unlock()
	w.ref.Tell(msg)
}

// 获取池大小
func (p *ActorPool) Len() int { p.mu.Lock(); defer p.mu.Unlock(); return len(p.workers) }

// 更新池大小指标
func (p *ActorPool) updatePoolMetrics() { actorPoolSize.WithLabelValues(p.name).Set(float64(p.Len())) }

// 启动自动扩缩容
func (p *ActorPool) StartAutoScaler(cfg AutoScalerConfig) {
	p.scaling.cfg = cfg
	p.scaling.stop = make(chan struct{})
	p.scaling.cool = time.Now()
	go func() {
		ticker := time.NewTicker(cfg.Interval)
		defer ticker.Stop()
		for {
			select {
			case <-ticker.C:
				p.mu.Lock()
				if time.Since(p.scaling.cool) < cfg.Cooldown {
					p.mu.Unlock()
					continue // 冷却期不扩缩容
				}
				total := 0
				for _, w := range p.workers {
					total += w.pending()
				}
				avg := 0
				if len(p.workers) > 0 {
					avg = total / len(p.workers)
				}
				p.mu.Unlock()
				if avg > cfg.HighThreshold {
					p.Resize(p.Len() + cfg.ScaleUpStep)
					p.scaling.cool = time.Now()
				}
				if avg < cfg.LowThreshold {
					p.Resize(p.Len() - cfg.ScaleDownStep)
					p.scaling.cool = time.Now()
				}
			case <-p.scaling.stop:
				fmt.Printf("[AUTOSCALE] %s 自动扩缩容已停止\n", p.name)
				return
			}
		}
	}()
}

// 停止自动扩缩容
func (p *ActorPool) StopAutoScaler() {
	if p.scaling.stop != nil {
		close(p.scaling.stop)
	}
}
func (p *ActorPool) Shutdown(ctx context.Context) error { p.StopAutoScaler(); return nil }

// ---------------- Main Example ----------------

// 主函数，演示 Actor 框架用法
func main() {
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	// 启动 Prometheus HTTP 指标服务
	go func() { http.Handle("/metrics", promhttp.Handler()); http.ListenAndServe(":2112", nil) }()

	// 创建 Actor 系统
	sys := newSystem()
	workerProps := Props{
		ActorFunc: func(ctx Context, msg Message) {
			switch m := msg.(type) {
			case int:
				time.Sleep(50 * time.Millisecond)
				fmt.Printf("[%s] 处理 %d\n", ctx.Self.Path(), m)
			case string:
				if m == "fail" {
					panic("模拟故障")
				}
			}
		},
		Mailbox: 16, Strategy: RestartOnFailure, MaxRestarts: 3, Window: 5 * time.Second,
	}

	// 创建 Actor 池
	pool := NewPool(sys, "testpool", workerProps, 2, 6)
	// 自动扩缩容配置
	autoCfg := AutoScalerConfig{Interval: 500 * time.Millisecond, HighThreshold: 6, LowThreshold: 2, ScaleUpStep: 2, ScaleDownStep: 1, Cooldown: 2 * time.Second}
	pool.StartAutoScaler(autoCfg)

	// 工作负载生成器，持续发送消息
	go func() {
		i := 0
		for {
			select {
			case <-ctx.Done():
				return
			default:
				if i%20 == 0 && i > 0 {
					pool.Tell("fail") // 每 20 次模拟故障
				} else {
					pool.Tell(i)
				}
				i++
				time.Sleep(50 * time.Millisecond)
			}
		}
	}()

	// 指标定时更新器
	go func() {
		ticker := time.NewTicker(500 * time.Millisecond)
		defer ticker.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				pool.mu.Lock()
				for _, w := range pool.workers {
					w.updateMetrics()
				}
				pool.updatePoolMetrics()
				pool.mu.Unlock()
			}
		}
	}()

	fmt.Println("Actor 系统运行中，按 Ctrl+C 退出。")
	<-ctx.Done()
	fmt.Println("\n[MAIN] 收到关闭信号，开始优雅关闭...")
	shutdownCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
	defer cancel()
	pool.StopAutoScaler()
	err := sys.Shutdown(shutdownCtx)
	if err != nil {
		fmt.Printf("[MAIN] 关闭超时: %v\n", err)
	} else {
		fmt.Println("[MAIN] 优雅关闭完成")
	}
	fmt.Println("系统退出")
}
